package com.dap.dubbo.rpc.filter;

import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.extension.Activate;
import com.alibaba.dubbo.common.utils.StringUtils;
import com.alibaba.dubbo.rpc.*;
import com.dap.dubbo.rpc.config.RedisConfig;
import org.redisson.api.RLock;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Transaction;
import redis.clients.jedis.ZParams;

import java.io.IOException;
import java.io.InputStream;
import java.util.*;

/*@Activate(group = {Constants.CONSUMER})*/
public class FlowControlFilter implements Filter {

    private long timeInterval_min = 60 * 1000;
    private long timeInterval_hour = 60 * 60 * 1000;
    private long timeInterval_day = 24 * 60 * 60 * 1000;
    private long timeInterval_sec = 1000;


    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {



        String resourceName = getResourceName(invoker, invocation);
        String application_group = invoker.getUrl().getParameter("group");
        String application = "FOP:FlowControl" + application_group;

        List<String> parameterList = parameterHandle(invoker);

        //每分钟频次
        String per_min_freq_num = parameterList.get(0);
        //每小时频次
        String per_hour_freq_num = parameterList.get(1);
        //每天时频次
        String per_day_freq_num = parameterList.get(2);
        //超限是否排队
        String queue_flag = parameterList.get(3);
        //队列等待超时
        String queue_timeout = parameterList.get(4);
        //队列最大长度
        String queue_max_length = parameterList.get(5);
        //每秒并发数
        String concurrent_num = parameterList.get(6);
        Boolean flag = judgeRank(resourceName, queue_timeout, queue_max_length);
        long sec_count = countServiceCall(timeInterval_sec, resourceName);
        long min_count = countServiceCall(timeInterval_min, resourceName);
        long hour_count = countServiceCall(timeInterval_hour, resourceName);
        long day_count = countServiceCall(timeInterval_day, resourceName);

        //拿到jedis
        Result invoke = null;
        Jedis jedis = RedisConfig.getJedis();
        try {

            Long now =System.currentTimeMillis();



            //(调试代码)将服务信息存入redis，如果有重复，则自动覆盖
            /*jedis.set(resourceName,"a");*/

            //获取令牌
            String identifier = UUID.randomUUID().toString();

            //超出限制范围
            if (sec_count > Long.parseLong(concurrent_num) || min_count > Long.parseLong(per_min_freq_num) ||
                    hour_count > Long.parseLong(per_hour_freq_num) || day_count > Long.parseLong(per_day_freq_num)) {
                //排队
                if (Boolean.parseBoolean(queue_flag)) {

                    if (flag) {//长度超限
                        return new RpcResult(new Exception("拒绝服务"));
                    } else {
                        //便于统计正常执行的请求数目
                        System.out.println("*****当前时间毫秒值："+now);
                        Transaction a = jedis.multi();
                        a.zadd("success" + resourceName, now, identifier);
                        a.exec();

                        invoke = invoker.invoke(invocation);
                    }

                } else {//不排队
                    /**
                     * 拒绝服务
                     */
                    return new RpcResult(new Exception("拒绝服务"));
                }

            } else {//正常范围内
                //便于统计正常执行的请求数目
                Transaction a = jedis.multi();
                a.zadd("success" + resourceName, now, identifier);
                a.exec();
                invoke = invoker.invoke(invocation);
            }
        } catch (NumberFormatException e) {
            e.printStackTrace();
        }  finally {
            jedis.close();
        }
        return invoke;
    }

    /**
     * 参数处理
     *
     * @param invoker
     * @return
     */
    private List<String> parameterHandle(Invoker<?> invoker) {

        Properties properties = null;
        String per_min_freq_num;
        String per_hour_freq_num;
        String per_day_freq_num;
        String queue_flag;
        String queue_timeout;
        String queue_max_length;
        String concurrent_num;

        String application_group = invoker.getUrl().getParameter("group");
        String application = "FOP:FlowControl_" + application_group;

        Jedis jedis = RedisConfig.getJedis();
        List<Object> results;
        try {
            Transaction multi = jedis.multi();
            multi.hget(application, "PER_MIN_FREQ_NUM");
            multi.hget(application, "PER_HOUR_FREQ_NUM");
            multi.hget(application, "PER_DAY_FREQ_NUM");
            multi.hget(application, "QUEUE_FLAG");
            multi.hget(application, "QUEUE_TIMEOUT");
            multi.hget(application, "QUEUE_MAX_LENGTH");
            multi.hget(application, "CONCURRENT_NUM");
            results = multi.exec();
        } finally {
            jedis.close();
        }
        boolean b = results.removeAll(Collections.singleton(null));

        if (b) {//如果有空值，则需要加载配置文件
            properties = new Properties();
            InputStream resourceAsStream = FlowControlFilter.class.getClassLoader().getResourceAsStream("properties/parameter.properties");
            try {
                properties.load(resourceAsStream);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        if (results.size() == 0 || results.get(0) == null || StringUtils.isEmpty(String.valueOf(results.get(0)))) {
            per_min_freq_num = properties.getProperty("per_min_freq_num", "1000");
        } else {
            per_min_freq_num = (String) results.get(0);
        }
        if (results.size() == 0 || results.get(1) == null || StringUtils.isEmpty(String.valueOf(results.get(1)))) {
            per_hour_freq_num = properties.getProperty("per_hour_freq_num", "1000");
        } else {
            per_hour_freq_num = (String) results.get(1);
        }
        if (results.size() == 0 || results.get(2) == null || StringUtils.isEmpty(String.valueOf(results.get(2)))) {
            per_day_freq_num = properties.getProperty("per_day_freq_num", "1000");
        } else {
            per_day_freq_num = (String) results.get(2);
        }
        if (results.size() == 0 || results.get(3) == null || StringUtils.isEmpty(String.valueOf(results.get(3)))) {
            queue_flag = properties.getProperty("queue_flag", "true");
        } else {
            queue_flag = (String) results.get(3);
        }
        if (results.size() == 0 || results.get(4) == null || StringUtils.isEmpty(String.valueOf(results.get(4)))) {
            queue_timeout = properties.getProperty("queue_timeout", "1000");
        } else {
            queue_timeout = (String) results.get(4);
        }
        if (results.size() == 0 || results.get(5) == null || StringUtils.isEmpty(String.valueOf(results.get(5)))) {
            queue_max_length = properties.getProperty("queue_max_length", "1000");
        } else {
            queue_max_length = (String) results.get(5);
        }
        if (results.size() == 0 || results.get(6) == null || StringUtils.isEmpty(String.valueOf(results.get(6)))) {
            concurrent_num = properties.getProperty("concurrent_num", "1000");
        } else {
            concurrent_num = (String) results.get(6);
        }

        List<String> parameterList = new ArrayList<String>();
        parameterList.add(per_min_freq_num);
        parameterList.add(per_hour_freq_num);
        parameterList.add(per_day_freq_num);
        parameterList.add(queue_flag);
        parameterList.add(queue_timeout);
        parameterList.add(queue_max_length);
        parameterList.add(concurrent_num);

        return parameterList;
    }

    /**
     * 判断是否超出队列最大长度
     *
     * @param resourceName     当前方法信息
     * @param queue_timeout    队列超时时间
     * @param queue_max_length 队列超时最大长度
     * @return false:未超出，true:超出
     */
    public boolean judgeRank(String resourceName, String queue_timeout, String queue_max_length) {

        //默认未超出
        boolean flag = false;


        //拿到jedis
        Jedis jedis = RedisConfig.getJedis();

        //获取令牌
        String identifier = UUID.randomUUID().toString();

        //当前时间毫秒值
        Long now = System.currentTimeMillis();
        Transaction transaction;
        Long rank = -1L;

        String lock = "lock" + resourceName;


        //得到锁
        RLock mylock = RedisConfig.getLock(lock);
        try {
            //加锁
            mylock.lock();
//            mylock.lock(2, TimeUnit.MINUTES); //lock提供带timeout参数，timeout结束强制解锁，防止死锁

            //调试代码
            System.out.println("==================lock======" + Thread.currentThread().getName());

            //开启redis事务
            transaction = jedis.multi();
            //返回移除的数量
            transaction.zremrangeByScore(("BUCKET_MONITOR" + resourceName).getBytes(), "-inf".getBytes(),
                    String.valueOf(now - Long.parseLong(queue_timeout)).getBytes());

            ZParams zParams = new ZParams();
            zParams.weightsByDouble(1.0, 0.0);

            //返回保存到目标结果的数量
            transaction.zinterstore("BUCKET" + resourceName, zParams, "BUCKET" + resourceName, "BUCKET_MONITOR" + resourceName);
            //服务访问总量
            transaction.incr("BUCKET" + resourceName + "count");
            List<Object> results = transaction.exec();

            //调试代码
            System.out.println("********************************调试代码result1=" + results);

            Long counter = (Long) results.get(results.size() - 1);

            //开启redis事务2
            Transaction transaction1 = jedis.multi();
            //添加时间元素
            transaction1.zadd("BUCKET_MONITOR" + resourceName, now, identifier);
            //添加计数元素
            transaction1.zadd("BUCKET" + resourceName, counter, identifier);
            //获取排序
            transaction1.zrank("BUCKET" + resourceName, identifier);
            results = transaction1.exec();

            //调试代码
            System.out.println("********************************调试代码result2=" + results);

            //排序 从1开始
            rank = (Long) results.get(results.size() - 1) + 1;

            //将当前请求在队列中的位置存入redis
            jedis.set("rank"+resourceName, String.valueOf(rank));

        } catch (Exception e) {
            e.printStackTrace();
        } finally {

            //调试代码
            System.out.println("==================解锁unlock======" + Thread.currentThread().getName());
            jedis.close();

            //解锁
            if (mylock.isLocked()) {
                mylock.unlock();
            }
        }


        //调试代码

        System.out.println("********************************调试代码rank=" + rank);
        //调试代码
        System.out.println("***********bucket的数量：" + jedis.zcount("BUCKET" + resourceName, "-inf", "+inf"));

        flag = rank > Long.parseLong(queue_max_length);

        //测试代码
        System.out.println("***********flag=" + flag + "*******rank=" + rank);

        //当前请求排序超过队列最大长度，则不加入队列
        if (flag) {
            Transaction transaction2 = jedis.multi();
            transaction2.zrem("BUCKET_MONITOR" + resourceName, identifier);
            transaction2.zrem("BUCKET" + resourceName, identifier);
            transaction2.exec();
        }

        return flag;

    }

    /**
     * 统计一段时间内的服务调用次数
     *
     * @param timeInterval 时间间隔
     * @param resourceName 服务方法信息
     * @return
     */
    public long countServiceCall(long timeInterval, String resourceName) {
        Long count = 0L;
        Jedis jedis = RedisConfig.getJedis();
        try {


            //获取令牌
            String identifier = UUID.randomUUID().toString();

            //当前时间毫秒值
            long now = System.currentTimeMillis();

            //开启redis事务
            Transaction transaction = jedis.multi();
            //给每个请求做标记
            transaction.zadd("count" + resourceName, now, identifier);
            //统计请求数量
            transaction.zcount("count" + resourceName, now - timeInterval, now);

            List<Object> results = transaction.exec();
            count = (Long) results.get(results.size() - 1);
        } finally {
            jedis.close();
        }

        //调试代码
        System.out.println("********************************调试代码+次数=" + count + "时间间隔：" + timeInterval);

        return count;

    }


    /**
     * 获取当前服务方法信息
     *
     * @param invoker
     * @param invocation
     * @return
     */
    private static String getResourceName(Invoker<?> invoker, Invocation invocation) {
        //服务名
        String interfaceName = invoker.getInterface().getName();
        //组名
        String application_group = invoker.getUrl().getParameter("group");
        String application = invoker.getUrl().getParameter("application");

        //拼接
        String resourceName =  "FOP:Application:"+application+"/Interface:"+interfaceName+"/Group:"+application_group;

        return resourceName;

    }
}
